[FLINK-37688] Implement Amazon CloudWatch Metric Sink Connector by darenwkt · Pull Request #202 · apache/flink-connector-aws

darenwkt · 2025-04-21T20:36:20Z

Purpose of the change

Implement Amazon CloudWatch Metric Sink Connector based on FLIP-524 (https://cwiki.apache.org/confluence/display/FLINK/FLIP-524:+CloudWatch+Metric+Sink+Connector) which includes following:

DataStream and TableAPI support
Integration Test with LocalStack container

Verifying this change

This change added tests and can be verified as follows:

Unit Test

Added unit tests

IT Test - Added integration tests for end-to-end deployment

Ran IT test locally

mvn clean verify -Prun-end-to-end-tests -DdistDir=/Users/darenwkt/Downloads/flink-1.19.2

Verified test pass for DataStream and TableAPI

[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 71.197 s - in org.apache.flink.connector.cloudwatch.table.test.CloudWatchTableAPIITCase
[INFO] Running org.apache.flink.connector.cloudwatch.sink.test.CloudWatchSinkITCase
[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 34.422 s - in org.apache.flink.connector.cloudwatch.sink.test.CloudWatchSinkITCase
[INFO] 
[INFO] Results:
[INFO] 
[INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0

DataStream API Manual Test - Manually verified by running the CloudWatch connector on a local Flink cluster.

Create Flink job code

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final Map<String, Properties> applicationParameters = loadApplicationProperties(env);

        env.fromSource(createDataGeneratorSource(applicationParameters.get("DataGen")), WatermarkStrategy.noWatermarks(), "DataGen")
                .uid("data-gen")
                .setParallelism(1)
                .disableChaining()
                .sinkTo(CloudWatchSink.<TemperatureSample>builder()
                        .setNamespace("CloudWatchSinkTest")
                        .setCloudWatchClientProperties(applicationParameters.get("CloudWatchSink"))
                        .setElementConverter(new MetricWriteRequestElementConverter<>() {
                            @Override
                            public MetricWriteRequest apply(TemperatureSample value, SinkWriter.Context context) {
                                return MetricWriteRequest.builder()
                                        .withMetricName("temperature")
                                        .addDimension("roomId", "testRoomId")
                                        .addDimension("sensorId", "testSensorId")
                                        .addValue(value.getTemperature())
                                        .addCount(1d)
                                        .withTimestamp(Instant.ofEpochMilli(value.getTimestamp()))
                                        .build();
                            }
                        })
                        .build())
                .uid("cloudwatch-sink").name("CloudWatchSink");

        env.enableCheckpointing(5000);

        env.execute("CloudWatch sink example");

Build jar and submitted job

./bin/flink run cloudwatch-connector-example-1.0.jar
Job has been submitted with JobID f417eccc9d9204604ec51eeed24196e0

Verified job is running and checkpointing successfully
Verified CloudWatch received the metric successfully

TableAPI Manual Test - Manually verified by running the CloudWatch connector on a local Flink cluster.

Built SQL jar and started sql-client with it

./bin/sql-client.sh --jar flink-sql-connector-cloudwatch-5.1-SNAPSHOT.jar

Created Sink Table

Flink SQL>
> CREATE TABLE CloudWatchTable
> (
>     `my_metric_name`   STRING,
>     `my_dim` STRING,
>     `sample_value` DOUBLE,
>     `sample_count` DOUBLE,
>     `unit`             STRING,
>     `storage_res`      INT,
>     `stats_max` DOUBLE,
>     `stats_min` DOUBLE,
>     `stats_sum` DOUBLE,
>     `stats_count` DOUBLE
> )
>     WITH (
>         'connector' = 'cloudwatch',
>         'aws.region' = 'us-east-1',
>         'metric.namespace' = 'cw_connector_namespace',
>         'metric.name.key' = 'my_metric_name',
>         'metric.dimension.keys' = 'my_dim',
>         'metric.value.key' = 'sample_value',
>         'metric.count.key' = 'sample_count',
>         'metric.unit.key' = 'unit',
>         'metric.storage-resolution.key' = 'storage_res',
>         'metric.statistic.max.key' = 'stats_max',
>         'metric.statistic.min.key' = 'stats_min',
>         'metric.statistic.sum.key' = 'stats_sum',
>         'metric.statistic.sample-count.key' = 'stats_count',
>         'sink.invalid-metric.retry-mode' = 'RETRY'
>         );
[INFO] Execute statement succeed.

Inserted sample value into table

Flink SQL> INSERT INTO CloudWatchTable VALUES ('test_metric', 'dim_1', 123, 1, 'Seconds', 60, 999, 1, 10, 1
1);
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 90d19a75e7326575b4ea3f4d883091a7

Verified job entered FINISHED state successfully
Verified Cloudwatch received the metric successfully

Significant changes

(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)

Dependencies have been added or upgraded
Public API has been changed (Public API is any class annotated with @Public(Evolving))
Serializers have been changed
New feature has been introduced
- If yes, how is this documented? (JavaDocs)

gguptp

Thanks for the PR @darenwkt! Left some comments

flink-connector-aws-e2e-tests/flink-connector-cloudwatch-e2e-tests/pom.xml

...-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/MetricWriteRequest.java

...va/org/apache/flink/connector/cloudwatch/sink/DefaultMetricWriteRequestElementConverter.java

gguptp · 2025-06-04T19:00:32Z

...in/java/org/apache/flink/connector/cloudwatch/sink/client/CloudWatchAsyncClientProvider.java

+    private final SdkAsyncHttpClient httpClient;
+    private final CloudWatchAsyncClient cloudWatchAsyncClient;


do these need to be class level variables? could we just expose the buildClient method instead. WDYT?

I see what you mean, I think the benefit of using getClient is that the same cloudwatchClient instance can be reused many times in the same provider, so I am inclined to keep this for now, let me know what you think

gguptp · 2025-06-04T19:01:14Z

...watch/src/main/java/org/apache/flink/connector/cloudwatch/sink/client/SdkClientProvider.java

+
+/** Provides a {@link SdkClient}. */
+@Internal
+public interface SdkClientProvider<T extends SdkClient> extends SdkAutoCloseable {


do we really need this interface?

We don't need it strictly, I just saw that we use this for DDB and SQS sink as well and thought of using it here to be consistent

...ctor-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/CloudWatchSink.java

flink-connector-aws/flink-sql-connector-cloudwatch/pom.xml

flink-connector-aws/flink-connector-cloudwatch/pom.xml

tcerda95 · 2025-08-10T13:06:12Z

Hey, this looks very good! What is the latest status on this? My team and I would be very keen to integrate a CloudWatch Table Sink, thank you very much for the contribution. Let me know if we can accelerate this PR. Cheers!

darenwkt · 2025-08-11T11:13:19Z

Hi @tcerda95 ! Great to hear from you and thanks for the interest! The PR is currently still being reviewed, and it would accelerate this PR if you could take a look at this PR too and leave any comments especially if you see any gap/differences with your team's requirements

@gguptp Thanks for the initial review on the PR and wondering if you would be able to take another look again please? Thank you

gguptp · 2025-09-11T15:10:33Z

Hi @tcerda95 ! Great to hear from you and thanks for the interest! The PR is currently still being reviewed, and it would accelerate this PR if you could take a look at this PR too and leave any comments especially if you see any gap/differences with your team's requirements

@gguptp Thanks for the initial review on the PR and wondering if you would be able to take another look again please? Thank you

@darenwkt will take a look by the end of the week

darenwkt force-pushed the cloudwatch branch 4 times, most recently from 2d0309c to 157fccd Compare April 25, 2025 19:31

darenwkt force-pushed the cloudwatch branch from 077dd35 to c543c17 Compare May 27, 2025 10:45

darenwkt closed this May 27, 2025

darenwkt force-pushed the cloudwatch branch from dfd1d9c to 9d6746b Compare May 27, 2025 10:47

darenwkt reopened this May 27, 2025

gguptp suggested changes Jun 4, 2025

View reviewed changes

darenwkt force-pushed the cloudwatch branch from a30b62b to 0797163 Compare June 8, 2025 07:43

darenwkt closed this Jun 8, 2025

darenwkt force-pushed the cloudwatch branch from 0797163 to 9d6746b Compare June 8, 2025 07:45

[FLINK-37688] Implement Amazon CloudWatch Metric Sink Connector

7d3ecf2

darenwkt reopened this Jun 8, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Comments

[FLINK-37688] Implement Amazon CloudWatch Metric Sink Connector#202

[FLINK-37688] Implement Amazon CloudWatch Metric Sink Connector#202
darenwkt wants to merge 1 commit intoapache:mainfrom
darenwkt:cloudwatch

darenwkt commented Apr 21, 2025 •

edited

Loading

Uh oh!

gguptp left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

gguptp Jun 4, 2025

Uh oh!

darenwkt Jun 8, 2025

Uh oh!

gguptp Jun 4, 2025

Uh oh!

darenwkt Jun 8, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

tcerda95 commented Aug 10, 2025

Uh oh!

darenwkt commented Aug 11, 2025

Uh oh!

gguptp commented Sep 11, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		private final SdkAsyncHttpClient httpClient;
		private final CloudWatchAsyncClient cloudWatchAsyncClient;

Comments

Conversation

darenwkt commented Apr 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose of the change

Verifying this change

Unit Test

IT Test - Added integration tests for end-to-end deployment

DataStream API Manual Test - Manually verified by running the CloudWatch connector on a local Flink cluster.

TableAPI Manual Test - Manually verified by running the CloudWatch connector on a local Flink cluster.

Significant changes

Uh oh!

gguptp left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

gguptp Jun 4, 2025

Choose a reason for hiding this comment

Uh oh!

darenwkt Jun 8, 2025

Choose a reason for hiding this comment

Uh oh!

gguptp Jun 4, 2025

Choose a reason for hiding this comment

Uh oh!

darenwkt Jun 8, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

tcerda95 commented Aug 10, 2025

Uh oh!

darenwkt commented Aug 11, 2025

Uh oh!

gguptp commented Sep 11, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

darenwkt commented Apr 21, 2025 •

edited

Loading